package com.algo.core.processor;

import com.algo.core.HttpRequestWrapper;
import com.algo.core.LifeCycle;
import com.algo.core.RapidSeverConfig;
import com.algo.core.filter.RapidFilterChain;
import org.apache.log4j.Logger;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 默认使用的请求处理器
 */
public class DefaultRequestProcessor implements RequestProcessor, LifeCycle {

    private final static Logger logger =org.apache.log4j.Logger.getLogger(DefaultRequestProcessor.class);

    // 配置类
    private RapidSeverConfig rapidSeverConfig;

    // 线程池
    private ExecutorService executorService;

    private PullRequestThread pullRequestThread;

    private RapidFilterChain filterChain;

    private volatile boolean isRunning = false;

    public DefaultRequestProcessor(RapidSeverConfig rapidSeverConfig, HttpRequestQueue httpRequestQueue,
                                   RapidFilterChain filterChain) {
        this.rapidSeverConfig = rapidSeverConfig;
        pullRequestThread = new PullRequestThread(httpRequestQueue);
        this.filterChain = filterChain;
        initilize();
    }

    @Override
    public void initilize() {
        // 初始化处理任务的线程池
        executorService = new ThreadPoolExecutor(Math.max(10, rapidSeverConfig.getRequestProcessorCoreThreadCount()),
                Math.max(10, rapidSeverConfig.getRequestProcessorMaxThreadCount()),
                rapidSeverConfig.getRequestProcessorThreadAliveTime(),
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                new ProcessorThreadFactory()
        );
    }

    @Override
    public void start() {
        isRunning = true;
        pullRequestThread.start();
    }

    @Override
    public void shutdown() {
        isRunning = false;
        // 关掉线程池
        executorService.shutdown();
    }

    @Override
    public void process(Runnable requstCommand) {
        // 提交到线程池异步处理
        executorService.submit(requstCommand);
    }

    /**
     * 将request queue中的请求读取出来，封装成一个一个可执行的command
     */
    private class PullRequestThread extends Thread {

        private HttpRequestQueue httpRequestQueue;

        public PullRequestThread(HttpRequestQueue httpRequestQueue) {
            this.httpRequestQueue = httpRequestQueue;
        }

        @Override
        public void run() {
            try {
                while (isRunning) {
                    HttpRequestWrapper httpWrapper = httpRequestQueue.take();
                    process(new RequestTask(httpWrapper, filterChain));
                }
            } catch (Exception e) {
                logger.error("PullRequestThread proccess error, ", e);
            }
        }
    }

}
